package com.atguigu.gmallrealtime.util;

import com.atguigu.gmallrealtime.common.Constant;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.JsonDebeziumSchemaSerializer;
import org.apache.doris.flink.sink.writer.SimpleStringSerializer;
import org.apache.flink.api.connector.sink.Sink;

import java.util.Properties;

/**
 * @author yhm
 * @create 2023-10-08 16:24
 */
public class DorisUtil {
    public static String getDorisSinkDDL(String tableName){
        return " WITH (\n" +
                "      'connector' = 'doris',\n" +
                "      'fenodes' = 'hadoop103:7030',\n" +
                "      'table.identifier' = 'gmall." +  tableName + "',\n" +
                "      'username' = '"  + Constant.DORIS_USER_NAME + "',\n" +
                "      'password' = '" + Constant.DORIS_PASSWORD + "',\n" +
                "      'sink.label-prefix' = 'gmall_label" + System.currentTimeMillis() +
                "')";
    }


    public static DorisSink<String> getDorisSink(String tableName) {
        Properties props = new Properties();
        props.setProperty("format", "json");
        props.setProperty("read_json_by_line", "true");

        DorisOptions dorisOptions = DorisOptions.builder()
                .setFenodes("hadoop102:7030")
                .setTableIdentifier("gmall." + tableName)
                .setUsername(Constant.DORIS_USER_NAME)
                .setPassword(Constant.DORIS_PASSWORD).build();

        DorisSink<String> dorisSink = DorisSink.<String>builder()
                .setDorisReadOptions(DorisReadOptions.builder().build())
                .setDorisExecutionOptions(DorisExecutionOptions.builder()
                        // 设置事务的label前缀
                        .setLabelPrefix("label-prefix"+ System.currentTimeMillis())
                        .disable2PC() // 开启两阶段提交后,labelPrefix 需要全局唯一,为了测试方便禁用两阶段提交
                        // 可以删除
                        .setDeletable(false)
                        .setBufferCount(3) // 批次条数: 默认 3
                        .setBufferSize(8 * 1024) // 批次大小: 默认 1M
                        .setCheckInterval(3000) // 批次输出间隔   三个对批次的限制是或的关系
                        .setMaxRetries(3)
                        // 设置写出数据的格式为json
                        .setStreamLoadProp(props)
                        .build())
                .setDorisOptions(dorisOptions)
                .setSerializer(new SimpleStringSerializer()).build();

        return dorisSink;
    }
}
